package com.atguigu.Flink.checkpoint;

import com.atguigu.Flink.function.ClickSource;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class Flink01_CheckPoint {
    public static void main(String[] args) {
        //设置系统变量
        System.setProperty("HADOOP_USER_NAME","atguigu") ;
        //在IDEA中 Edit Configurations...中添加 VM 参数
        //-DHADOOP_USER_NAME=atguigu

        Configuration conf = new Configuration();
        conf.setInteger("rest.port" , 5678);
        //最终检查点
        conf.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, false);


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);


        //1. 开启检查点
        env.enableCheckpointing(5000L , CheckpointingMode.EXACTLY_ONCE) ;  //默认精确一次


        //获取检查点配置对象
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();

        //2. 检查点存储位置
        //  JobManager的内存 :  JobManagerCheckpointStorage
        //  持久化存储系统中(例如:HDFS ) : FileSystemCheckpointStorage
        checkpointConfig.setCheckpointStorage( new Path("hdfs://hadoop102:8020/ck"));

        //3. 检查点超时时间
        checkpointConfig.setCheckpointTimeout(10000);

        //4. 同时存在的检查点个数
        checkpointConfig.setMaxConcurrentCheckpoints(1);

        //5. 两个检查点之间的间隔时间
        //checkpointConfig.setMinPauseBetweenCheckpoints(5000);

        //6. 检查点允许失败的次数
        checkpointConfig.setTolerableCheckpointFailureNumber(3);

        //7. 停掉作业后，外部的检查点数据是否要清理
        //   DELETE_ON_CANCELLATION :  清理
        //   RETAIN_ON_CANCELLATION :  保留
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        //8. 开启非对齐检查点策略
        // 注意: 同时存在的检查点个数不能大于1 。
        checkpointConfig.enableUnalignedCheckpoints();

        //9. 对齐检查点的超时时间
        // 注意: 对齐检查点如果超时，直接开启非对齐。
        checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(10));

        //10.配置RockDB状态后端， 并开启增量
        //在配置文件中配置: state.backend.incremental: true
        env.setStateBackend(new EmbeddedRocksDBStateBackend(true ));

        //11. 开启changlog
        env.enableChangelogStateBackend(true) ;

        //12. 最终检查点
        //Configuration conf = new Configuration();
        //conf.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, false);


        env.addSource(new ClickSource())
                .map( x -> x ).uid("m1")
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
